package com.qf.simple;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer_manual {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //rabbitmq服务器安装的机器IP
        factory.setHost("localhost");
        //rabbitmq通过amqp协议连接时的端口号
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        //设置消息数据保存的虚拟主机,不同的虚拟主机彼此隔离
        factory.setVirtualHost("cdm");

        //获取connection对象
        Connection connection = factory.newConnection();
        //通过connection获取内部通道
        Channel channel = connection.createChannel();

        //消费消息
        /*
            String queue: 队列名称
            boolean autoAck ：是否自动确认（一旦确认，消息就会删除；没有确认，消息不会删除--①死信队列②重新入队（重新消费））
            String consumerTag: 消费者标识，同一个消费者，标识唯一不变
            boolean noLocal：现在不支持
            boolean exclusive： 是否排他
            Map<String, Object> arguments： 额外参数
            DeliverCallback deliverCallback： 消息送达的回调
            CancelCallback cancelCallback： 消息取消的回调
            ConsumerShutdownSignalCallback shutdownSignalCallback： 消费者关闭的回调

            String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback -- 消息自动确认
         */
        channel.basicConsume("simple.que",false, new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                try {

                    //模拟失败
                    int i = 1/0;

                    System.out.println("当前消费者：" + consumerTag);
                    //消息内容
                    byte[] body = message.getBody();
                    System.out.println(new String(body, "utf-8"));

                    //封装消息交换机、路由键、队列信息的对象
                    Envelope envelope = message.getEnvelope();
                    //消息的额外参数
                    AMQP.BasicProperties properties = message.getProperties();

                    //手动确认
                /*
                    long deliveryTag: 生产端的唯一标识
                    boolean multiple：true，一次性将该生产端之前未消费的消息一并确认；false，只确认当前消费的这条消息
                 */
                    channel.basicAck(envelope.getDeliveryTag(), true);
                }catch (Exception e){
                    //消息消费失败了 -- 不能确认
                    /*
                        long deliveryTag
                        boolean multiple
                        boolean requeue: 是否重新入队；是-重新消费,否-丢弃（进入死信队列）
                            重新入队：
                                ① 可能存在幂等问题
                                ② 可能造成消费死循环，消费者服务的压力变大了
                                      一般我们在业务中不会无限入队，需要控制重新入队的次数。当达到次数后（5次），依然消费失败
                                        我们将这条消息写入数据库的重试表中，利用定时任务执行重新消费
                                      计数器：redis--- key:消息的id，value:重新入队的次数   -- incr
                                               value = jedis.incr(消息的id);
                     */
                    System.out.println("消费失败...");
                    channel.basicNack(message.getEnvelope().getDeliveryTag(),true,true);
                }
            }
        }, new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {
                System.out.println("消息取消:" + consumerTag);
            }
        });

        //消费者启动后，一直处于监听状态，资源不需要关闭
    }
}
